package com.zf.nospring.kafka.transaction;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 消费 trans_test 主题中的消息
 */
public class TransactionConsumerTest {

    @Test
    public void testConsumer() {
        // 创建消费者的配置对象
        Properties properties = new Properties();
        // kafka broker的地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 配置消费者组ID，任意起名，必传
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "dtrans");

        // 一次 poll 拉取，获取消息的最大条数，默认是 500 条。
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);

        // 开启自动提交offset功能，默认开启。offset从0开始
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 每隔多少毫秒自动提交一次 offset
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 50);

        // 设置隔离级别为 读取已提交的数据，默认是读未提交的数据。
        // 防止读取到事务没有提交的数据
        properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString().toLowerCase());

        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        // 注册要消费的主题（可以消费多个主题）
        try (kafkaConsumer) {
            kafkaConsumer.subscribe(Arrays.asList("trans_test"));
            while (true) {
                // 获取消息，超时时间设置为1s, 1s后如果没有获取到消息则返回空集
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
                // 打印消费到的数据
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    String topic = consumerRecord.topic();
                    String msg = consumerRecord.value();
                    long offset = consumerRecord.offset();
                    int partition = consumerRecord.partition();
                    System.out.printf("topic: %s, msg: %s, offset: %d, partition: %d\n", topic, msg, offset, partition);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
